// websocket通信协议在本项目中已经彻底废弃
// 所有连接改用http协议进行通信

use axum_extra::TypedHeader;
use serde::{Deserialize, Serialize};
use axum::{
    extract::{
        ws::{Message, WebSocket, WebSocketUpgrade},
        ConnectInfo,
    },
    response::IntoResponse,
};
use futures_util::{StreamExt, SinkExt};
use std::{net::SocketAddr, collections::HashMap};
use tokio::sync::{mpsc::{UnboundedSender, UnboundedReceiver, unbounded_channel}, Mutex};
use serde_json;
use std::sync::Arc;
use once_cell::sync::Lazy;
use headers::UserAgent;

// -------------------- 数据结构 --------------------

#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Config {
    pub key: String,
    pub value: String,
}

// -------------------- 全局发送器 --------------------
// 用于广播消息给所有连接的客户端

static CONFIG_DISPATCHER: Lazy<Arc<Mutex<HashMap<SocketAddr, UnboundedSender<Config>>>>> =
    Lazy::new(|| Arc::new(Mutex::new(HashMap::new())));

pub async fn send_config(config: Config) {
    let dispatcher = CONFIG_DISPATCHER.lock().await;
    for (addr, sender) in dispatcher.iter() {
        if let Err(e) = sender.send(config.clone()) {
            println!("❌ 广播给 {addr} 失败: {e}");
        }
    }
}

// -------------------- WebSocket 接入 --------------------

pub async fn ws_handler(
    ws: WebSocketUpgrade,
    user_agent: Option<TypedHeader<UserAgent>>,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
) -> impl IntoResponse {
    let user_agent = user_agent
        .map(|TypedHeader(ua)| ua.to_string())
        .unwrap_or_else(|| "Unknown browser".to_string());

    println!("✅ `{user_agent}` at {addr} connected.");
    ws.on_upgrade(move |socket| handle_socket(socket, addr))
}

// -------------------- WebSocket 主处理 --------------------

async fn handle_socket(socket: WebSocket, who: SocketAddr) {
    println!("🔗 WebSocket 连接建立：{who}");

    let (mut ws_sender, mut ws_receiver) = socket.split();
    let (config_tx, mut config_rx): (UnboundedSender<Config>, UnboundedReceiver<Config>) = unbounded_channel();

    {
        let mut dispatcher = CONFIG_DISPATCHER.lock().await;
        dispatcher.insert(who, config_tx);
    }

    // 接收前端消息
    let recv_task = tokio::spawn(async move {
        while let Some(result) = ws_receiver.next().await {
            match result {
                Ok(msg) => {
                    if let Message::Close(_) = msg {
                        println!("❎ 客户端 {who} 主动关闭连接");
                        break;
                    }

                    if let Some(config) = parse_config_message(msg, who) {
                        println!("📥 收到来自 {who} 的 Config：key = {}, value = {}", config.key, config.value);
                        handle_config(config, who).await;
                    }
                }
                Err(e) => {
                    println!("❌ 客户端 {who} 异常断开: {e}");
                    break;
                }
            }
        }
    });

    // 发送后端消息
    let send_task = tokio::spawn(async move {
        while let Some(config) = config_rx.recv().await {
            match serde_json::to_string(&config) {
                Ok(json) => {
                    if ws_sender.send(Message::Text(json.into())).await.is_err() {
                        println!("❌ 发送给 {who} 失败，连接可能已断开");
                        break;
                    }
                }
                Err(e) => {
                    println!("❌ 序列化 Config 失败: {e}");
                }
            }
        }
    });

    tokio::select! {
        _ = recv_task => {},
        _ = send_task => {},
    }

    {
        let mut dispatcher = CONFIG_DISPATCHER.lock().await;
        dispatcher.remove(&who);
    }

    println!("🔚 连接 {who} 已关闭");
}

// -------------------- 消息解析 --------------------

fn parse_config_message(msg: Message, who: SocketAddr) -> Option<Config> {
    match msg {
        Message::Text(text) => {
            match serde_json::from_str::<Config>(&text) {
                Ok(config) => Some(config),
                Err(e) => {
                    println!("❌ 无法解析来自 {who} 的 JSON: {e}");
                    None
                }
            }
        }
        Message::Binary(_) => {
            println!("⚠️ 收到来自 {who} 的二进制消息，但不支持");
            None
        }
        Message::Close(c) => {
            println!("📴 收到来自 {who} 的关闭请求: {:?}", c);
            None
        }
        Message::Ping(v) => {
            println!("📡 收到来自 {who} 的 ping: {:?}", v);
            None
        }
        Message::Pong(v) => {
            println!("📡 收到来自 {who} 的 pong: {:?}", v);
            None
        }
    }
}

// -------------------- 后端处理逻辑 --------------------

async fn handle_config(config: Config, who: SocketAddr) {
    println!("🛠️ 处理来自 {who} 的配置：{:?}", config);
    // 这里可以执行数据库写入、状态更新等操作

    match config.key.as_str() {
        // 处理commands指令
        "commands" => {

            match config.value.as_str() {
                "get_console" => {
                    get_console_ws().await;
                }
                "get_volume" => {
                    _get_volume().await;
                }
                "get_video_code_rate" => {
                    _get_video_code_rate().await;
                }
                "get_audio_code_rate" => {
                    _get_audio_code_rate().await;
                }
                "get_video_available" => {
                    _get_video_available().await;
                }
                "get_audio_available" => {
                    _get_audio_available().await;
                }
                "get_turn" => {
                    _get_turn().await;
                }
                // 改用http协议, 此处作废
                // "stream_start" => {
                //     _stream_start().await;
                // }
                // "close" => {
                //     _close().await;
                // }
                "db_delete_all" => {
                    _db_delete_all().await;
                }

                // 处理commands未匹配指令
                _ => {
                    eprintln!("⚠️ 收到来自 {who} 的未知指令：{:?}", config);
                }

            }
        }

        // 处理非commands指令
        _ => {
            match config.key.as_str() {
                "console_id" => {
                    save_console_id(&config.value).await;
                }
                "volume" => {
                    save_volume(&config.value).await;
                }
                "video_code_rate" => {
                    save_video_code_rate(&config.value).await;
                }
                "audio_code_rate" => {
                    save_audio_code_rate(&config.value).await;
                }
                "video_available" => {
                    save_video_available(&config.value).await;
                }
                "audio_available" => {
                    save_audio_available(&config.value).await;
                }
                "turn_server" => {
                    save_turn_server(&config.value).await;
                }

                // 处理未匹配指令
                _ => {
                    eprintln!("⚠️ 收到来自 {who} 的未知指令：{:?}", config);
                }
            }
        }
    }
    // 暂时将所有发送的所有数据进行原地返回
    send_config(config).await;
}


// -------------------- 存放commands指令响应函数 --------------------
// 返回主机列表的json示例
// value里面是字符串, 为了方便直观的查看, 这里的value显示的不是字符串
// 但是在实际的程序中, value应该是一个字符串
// 实际正确的json在下面
// 注意注意注意: 这里的value是字符串, 不是json, 实际值为json字符串
// {
//   "key": "console_list",
//   "value": {
//     "console": [
//       {
//         "id": "F4001D0F950AA28A",
//         "name": "客厅 Xbox",
//         "locale": "zh-Hans-SG",
//         "powerState": "ConnectedStandby",
//         "consoleType": "XboxSeriesS"
//       },
//       {
//         "id": "B123456789ABCDEF",
//         "name": "卧室 Xbox",
//         "locale": "en-US",
//         "powerState": "On",
//         "consoleType": "XboxSeriesX"
//       }
//     ]
//   }
// }
// 
// 正确的json示例
// {
//   "key": "console_list",
//   "value": "{\"console\":[{\"id\":\"F4001D0F950AA28A\",\"name\":\"客厅 Xbox\",\"locale\":\"zh-Hans-SG\",\"powerState\":\"ConnectedStandby\",\"consoleType\":\"XboxSeriesS\"}]}"
// }
use crate::api::get_console::{get_console, Console};
#[derive(Debug, Serialize, Deserialize)]
struct ConsoleList {
    console: Vec<Console>,
}
async fn get_console_ws() {
    let console = get_console().await.unwrap();
    let console_list = ConsoleList { console };
    let console_str = serde_json::to_string(&console_list).unwrap();
    send_config(Config {
        key: "console_list".to_string(),
        value: console_str,
    }).await;
}
use crate::media::volume::get_volume;
async fn _get_volume() {
    let volume = get_volume().await;
    send_config(Config 
        {
            key: "volume".to_string(),
            value: volume.to_string()
        }
    ).await;
}
// 单位: kbps
use crate::web_rtc::code_rate::get_video_code_rate;
async fn _get_video_code_rate() {
    let video_code_rate = get_video_code_rate().await;
    send_config(Config {
        key: "video_code_rate".to_string(),
        value: video_code_rate.to_string(),
    }).await;
}
// 单位: kbps
use crate::web_rtc::code_rate::get_audio_code_rate;
async fn _get_audio_code_rate() {
    let audio_code_rate = get_audio_code_rate().await;
    send_config(Config {
        key: "audio_code_rate".to_string(),
        value: audio_code_rate.to_string(),
    }).await;
}
use crate::media::media_ctrl::get_video_available;
async fn _get_video_available() {
    let video_available = get_video_available().await;
    send_config(Config {
        key: "video_available".to_string(),
        value: video_available.to_string()
    }).await;
}
use crate::media::media_ctrl::get_audio_available;
async fn _get_audio_available() {
    let audio_available = get_audio_available().await;
    send_config(Config {
        key: "audio_available".to_string(),
        value: audio_available.to_string()
    }).await;
}

#[derive(Debug, Serialize, Deserialize)]
struct turn_server {
    server: String,
    username: String,
    password: String,
}
/**
 * 通信格式:
 * 发送和接收都是这个格式
 * {
 *   "key": "turn_server",
 *   "value": "{\"server\":\"turn:turn.xlink.cn:3478\",\"username\":\"xlink\",\"password\":\"xlink\"}"
 * }
 */
use crate::web_rtc::turn::TurnServerConfig;
async fn _get_turn() {
    if let Some(turn) = TurnServerConfig::get().await {
        let _turn_server = turn_server {
            server: turn.server,
            username: turn.username,
            password: turn.password,
        };
        let turn_server_str = serde_json::to_string(&_turn_server).unwrap();
        send_config(Config {
            key: "turn_server".to_string(),
            value: turn_server_str,
        }).await;
    } else {
        send_config(Config {
            key: "turn_server".to_string(),
            value: "null".to_string(),
        }).await;
    }
}
// use crate::web_rtc::init::init;
// async fn _stream_start() {
//     init().await.unwrap();
// }
// use crate::web_rtc::close::close;
// async fn _close() {
//     close().await;
// }
use crate::db::db_delete::db_delete_all;
async fn _db_delete_all() {
    db_delete_all().await.unwrap();
}
// -------------------- 存放非commands指令响应函数 --------------------
use crate::api::get_session_id::get_session_id;
async fn save_console_id(console_id: &str) {
    get_session_id(Some(console_id.to_string())).await.unwrap();
}
use crate::media::volume::set_volume;
async fn save_volume(volume: &str) {
    set_volume(volume.parse::<u8>().unwrap()).await;
}
use crate::web_rtc::code_rate::set_video_code_rate;
async fn save_video_code_rate(video_code_rate: &str) {
    set_video_code_rate(video_code_rate.parse::<u32>().unwrap()).await;
}
use crate::web_rtc::code_rate::set_audio_code_rate;
async fn save_audio_code_rate(audio_code_rate: &str) {
    set_audio_code_rate(audio_code_rate.parse::<u32>().unwrap()).await;
}
use crate::media::media_ctrl::set_video_available;
async fn save_video_available(video_available: &str) {
    set_video_available(video_available.parse::<bool>().unwrap()).await;
}
use crate::media::media_ctrl::set_audio_available;
async fn save_audio_available(audio_available: &str) {
    set_audio_available(audio_available.parse::<bool>().unwrap()).await;
}
/**
 * 通信格式:
 * 发送和接收都是这个格式
 * {
 *   "key": "turn_server",
 *   "value": "{\"server\":\"turn:turn.xlink.cn:3478\",\"username\":\"xlink\",\"password\":\"xlink\"}"
 * }
 */
async fn save_turn_server(turn_server: &str) {
    let turn_server: turn_server = serde_json::from_str(turn_server).unwrap();
    TurnServerConfig::set(&turn_server.server, &turn_server.username, &turn_server.password).await;
}
